package com.源码分析.优化;

import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.CoreSubscriber;
import reactor.core.publisher.Flux;

public class Demo1 {

	private static Logger log = LoggerFactory.getLogger(Demo1.class);

	public static void main(String[] args) throws InterruptedException {
		Flux.just(1, 2, 3)
			.filter(x -> x > 2)
			.subscribe(new CoreSubscriber<Integer>() {// 这里传入CoreSubscriber对象作为订阅者
				private Subscription subscription;
				@Override
				public void onSubscribe(Subscription s) {
					// 保存订阅关系, 需要用它来给发布者响应
	                this.subscription = s;
					log.info("onSubscribe, {}", s.getClass());
					// 请求5个数据
					s.request(1);
				}

				@Override
				public void onNext(Integer integer) {
					log.info("onNext： {}", integer);
					// 处理完调用request再请求一个数据
	                this.subscription.request(1);
	                // 或者 已经达到了目标, 调用cancel告诉发布者不再接受数据了
	                // this.subscription.cancel();
				}

				@Override
				public void onError(Throwable t) {
					// 出现了异常(例如处理数据的时候产生了异常)
	                t.printStackTrace();

					// 我们可以告诉发布者, 后面不接受数据了
	                this.subscription.cancel();
				}

				@Override
				public void onComplete() {
					// 全部数据处理完了(发布者关闭了)
					log.info("onComplete");
				}
			});
		Thread.sleep(1000);
	}

}
